

/*-------------------------------------------------------------------------
 *
 * heap_surgery.c
 *	  对损坏的堆表进行手术的功能。
 *
 * Copyright (c) 2020-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  contrib/pg_surgery/heap_surgery.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/heapam.h"
#include "access/visibilitymap.h"
#include "access/xloginsert.h"
#include "catalog/pg_am_d.h"
#include "catalog/pg_proc_d.h"
#include "miscadmin.h"
#include "storage/bufmgr.h"
#include "utils/acl.h"
#include "utils/rel.h"

PG_MODULE_MAGIC;

/* 强制改变堆元组状态的选项。 */
typedef enum HeapTupleForceOption
{
	HEAP_FORCE_KILL,
	HEAP_FORCE_FREEZE
} HeapTupleForceOption;

PG_FUNCTION_INFO_V1(heap_force_kill);
PG_FUNCTION_INFO_V1(heap_force_freeze);

static int32 fc_tidcmp(const void *fc_a, const void *fc_b);
static Datum fc_heap_force_common(FunctionCallInfo fcinfo,
							   HeapTupleForceOption fc_heap_force_opt);
static void fc_sanity_check_tid_array(ArrayType *fc_ta, int *fc_ntids);
static BlockNumber fc_find_tids_one_page(ItemPointer fc_tids, int fc_ntids,
									  OffsetNumber *fc_next_start_ptr);

/*-------------------------------------------------------------------------
 * heap_force_kill()
 *
 * 强制删除由给定 TID 数组中存储的项指针指向的元组。
 *
 * 用法：SELECT heap_force_kill(regclass, tid[]);
 *-------------------------------------------------------------------------
 */
Datum heap_force_kill(PG_FUNCTION_ARGS)
{
	PG_RETURN_DATUM(fc_heap_force_common(fcinfo, HEAP_FORCE_KILL));
}

/*-------------------------------------------------------------------------
 * heap_force_freeze()
 *
 * 强制冻结由给定 TID 数组中存储的项指针指向的元组。
 *
 * 用法：SELECT heap_force_freeze(regclass, tid[]);
 *-------------------------------------------------------------------------
 */
Datum heap_force_freeze(PG_FUNCTION_ARGS)
{
	PG_RETURN_DATUM(fc_heap_force_common(fcinfo, HEAP_FORCE_FREEZE));
}

/*-------------------------------------------------------------------------
 * heap_force_common()
 *
 * heap_force_kill 和 heap_force_freeze 的通用代码
 *-------------------------------------------------------------------------
 */
static Datum fc_heap_force_common(FunctionCallInfo fcinfo, HeapTupleForceOption fc_heap_force_opt)
{
	Oid			fc_relid = PG_GETARG_OID(0);
	ArrayType  *fc_ta = PG_GETARG_ARRAYTYPE_P_COPY(1);
	ItemPointer fc_tids;
	int			fc_ntids,
				fc_nblocks;
	Relation	fc_rel;
	OffsetNumber fc_curr_start_ptr,
				fc_next_start_ptr;
	bool		fc_include_this_tid[MaxHeapTuplesPerPage];

	if (RecoveryInProgress())
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("recovery is in progress"),
				 errhint("heap surgery functions cannot be executed during recovery.")));

	/* 检查输入。 */
	fc_sanity_check_tid_array(fc_ta, &fc_ntids);

	fc_rel = relation_open(fc_relid, RowExclusiveLock);

	/*
	 * 检查目标关系。
	 */
	if (!RELKIND_HAS_TABLE_AM(fc_rel->rd_rel->relkind))
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("cannot operate on relation \"%s\"",
						RelationGetRelationName(fc_rel)),
				 errdetail_relkind_not_supported(fc_rel->rd_rel->relkind)));

	if (fc_rel->rd_rel->relam != HEAP_TABLE_AM_OID)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("only heap AM is supported")));

	/* 必须是表的拥有者或超级用户。 */
	if (!pg_class_ownercheck(RelationGetRelid(fc_rel), GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER,
					   get_relkind_objtype(fc_rel->rd_rel->relkind),
					   RelationGetRelationName(fc_rel));

	fc_tids = ((ItemPointer) ARR_DATA_PTR(fc_ta));

	/*
	 * 如果数组中有多个 TID，则对其进行排序，以便我们能够
	 * 轻松地从数组中提取属于某特定页面的所有 TIDs。
	 */
	if (fc_ntids > 1)
		qsort((void *) fc_tids, fc_ntids, sizeof(ItemPointerData), fc_tidcmp);

	fc_curr_start_ptr = fc_next_start_ptr = 0;
	fc_nblocks = RelationGetNumberOfBlocks(fc_rel);

	/*
	 * 循环，为每个块执行必要的操作。
	 */
	while (fc_next_start_ptr != fc_ntids)
	{
		Buffer		fc_buf;
		Buffer		fc_vmbuf = InvalidBuffer;
		Page		fc_page;
		BlockNumber fc_blkno;
		OffsetNumber fc_curoff;
		OffsetNumber fc_maxoffset;
		int			fc_i;
		bool		fc_did_modify_page = false;
		bool		fc_did_modify_vm = false;

		CHECK_FOR_INTERRUPTS();

		/*
		 * 从 next_start_ptr 开始查找属于某特定页面的所有 TID，
		 * 并逐个处理它们。
		 */
		fc_blkno = fc_find_tids_one_page(fc_tids, fc_ntids, &fc_next_start_ptr);

		/* 检查块编号是否有效。 */
		if (fc_blkno >= fc_nblocks)
		{
			/* 在移动到下一个页面之前更新 current_start_ptr。 */
			fc_curr_start_ptr = fc_next_start_ptr;

			ereport(NOTICE,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("skipping block %u for relation \"%s\" because the block number is out of range",
							fc_blkno, RelationGetRelationName(fc_rel))));
			continue;
		}

		fc_buf = ReadBuffer(fc_rel, fc_blkno);
		LockBufferForCleanup(fc_buf);

		fc_page = BufferGetPage(fc_buf);

		fc_maxoffset = PageGetMaxOffsetNumber(fc_page);

		/*
		 * 确定我们将要处理哪些 TIDs 和哪些 TIDs 将被跳过。
		 */
		memset(fc_include_this_tid, 0, sizeof(fc_include_this_tid));
		for (fc_i = fc_curr_start_ptr; fc_i < fc_next_start_ptr; fc_i++)
		{
			OffsetNumber fc_offno = ItemPointerGetOffsetNumberNoCheck(&fc_tids[fc_i]);
			ItemId		fc_itemid;

			/* 检查偏移编号是否有效。 */
			if (fc_offno == InvalidOffsetNumber || fc_offno > fc_maxoffset)
			{
				ereport(NOTICE,
						errmsg("skipping tid (%u, %u) for relation \"%s\" because the item number is out of range",
							   fc_blkno, fc_offno, RelationGetRelationName(fc_rel)));
				continue;
			}

			fc_itemid = PageGetItemId(fc_page, fc_offno);

			/* 仅接受使用过的项 ID。 */
			if (ItemIdIsRedirected(fc_itemid))
			{
				ereport(NOTICE,
						errmsg("skipping tid (%u, %u) for relation \"%s\" because it redirects to item %u",
							   fc_blkno, fc_offno, RelationGetRelationName(fc_rel),
							   ItemIdGetRedirect(fc_itemid)));
				continue;
			}
			else if (ItemIdIsDead(fc_itemid))
			{
				ereport(NOTICE,
						(errmsg("skipping tid (%u, %u) for relation \"%s\" because it is marked dead",
								fc_blkno, fc_offno, RelationGetRelationName(fc_rel))));
				continue;
			}
			else if (!ItemIdIsUsed(fc_itemid))
			{
				ereport(NOTICE,
						(errmsg("skipping tid (%u, %u) for relation \"%s\" because it is marked unused",
								fc_blkno, fc_offno, RelationGetRelationName(fc_rel))));
				continue;
			}

			/* 标记为待处理。 */
			Assert(fc_offno < MaxHeapTuplesPerPage);
			fc_include_this_tid[fc_offno] = true;
		}

		/*
		 * 在进入临界区之前，如果有必要，请固定可见性映射页面。
		 */
		if (fc_heap_force_opt == HEAP_FORCE_KILL && PageIsAllVisible(fc_page))
			visibilitymap_pin(fc_rel, fc_blkno, &fc_vmbuf);

		/* 从这里开始，直到所有更改记录为止，不允许有 ereport(ERROR)。 */
		START_CRIT_SECTION();

		for (fc_curoff = FirstOffsetNumber; fc_curoff <= fc_maxoffset;
			 fc_curoff = OffsetNumberNext(fc_curoff))
		{
			ItemId		fc_itemid;

			if (!fc_include_this_tid[fc_curoff])
				continue;

			fc_itemid = PageGetItemId(fc_page, fc_curoff);
			Assert(ItemIdIsNormal(fc_itemid));

			fc_did_modify_page = true;

			if (fc_heap_force_opt == HEAP_FORCE_KILL)
			{
				ItemIdSetDead(fc_itemid);

				/*
				 * 如果页面标记为全部可见，我们必须清除
				 * 页面头中的 PD_ALL_VISIBLE 标志，以及
				 * 对应于该页面的可见性映射中的全部可见位。
				 */
				if (PageIsAllVisible(fc_page))
				{
					PageClearAllVisible(fc_page);
					visibilitymap_clear(fc_rel, fc_blkno, fc_vmbuf,
										VISIBILITYMAP_VALID_BITS);
					fc_did_modify_vm = true;
				}
			}
			else
			{
				HeapTupleHeader fc_htup;

				Assert(fc_heap_force_opt == HEAP_FORCE_FREEZE);

				fc_htup = (HeapTupleHeader) PageGetItem(fc_page, fc_itemid);

				/*
				 * 重置元组的所有可见性相关字段。此
				 * 逻辑应模仿 heap_execute_freeze_tuple()，
				 * 但我们选择重置 xmin 和 ctid，以确保没有
				 * 潜在的乱数据被遗留。
				 */
				ItemPointerSet(&fc_htup->t_ctid, fc_blkno, fc_curoff);
				HeapTupleHeaderSetXmin(fc_htup, FrozenTransactionId);
				HeapTupleHeaderSetXmax(fc_htup, InvalidTransactionId);
				if (fc_htup->t_infomask & HEAP_MOVED)
				{
					if (fc_htup->t_infomask & HEAP_MOVED_OFF)
						HeapTupleHeaderSetXvac(fc_htup, InvalidTransactionId);
					else
						HeapTupleHeaderSetXvac(fc_htup, FrozenTransactionId);
				}

				/*
				 * 清除该元组的所有可见性相关位并
				 * 将其标记为冻结。同时，清除 HOT_UPDATED 和
				 * KEYS_UPDATES 位。
				 */
				fc_htup->t_infomask &= ~HEAP_XACT_MASK;
				fc_htup->t_infomask |= (HEAP_XMIN_FROZEN | HEAP_XMAX_INVALID);
				fc_htup->t_infomask2 &= ~HEAP_HOT_UPDATED;
				fc_htup->t_infomask2 &= ~HEAP_KEYS_UPDATED;
			}
		}

		/*
		 * 如果页面被修改，只有在这种情况下，我们才标记缓冲区为脏或执行
		 * WAL 记录。
		 */
		if (fc_did_modify_page)
		{
			/* 在写入 WAL 之前标记缓冲区为脏。 */
			MarkBufferDirty(fc_buf);

			/* XLOG 相关内容 */
			if (RelationNeedsWAL(fc_rel))
				log_newpage_buffer(fc_buf, true);
		}

		/* 如果 VM 页面被修改，则记录 WAL。 */
		if (fc_did_modify_vm && RelationNeedsWAL(fc_rel))
			log_newpage_buffer(fc_vmbuf, false);

		END_CRIT_SECTION();

		UnlockReleaseBuffer(fc_buf);

		if (fc_vmbuf != InvalidBuffer)
			ReleaseBuffer(fc_vmbuf);

		/* 在移动到下一个页面之前更新 current_start_ptr。 */
		fc_curr_start_ptr = fc_next_start_ptr;
	}

	relation_close(fc_rel, RowExclusiveLock);

	pfree(fc_ta);

	PG_RETURN_VOID();
}

/*-------------------------------------------------------------------------
 * tidcmp()
 *
 * 比较两个项指针，返回 -1, 0 或 +1。
 *
 * 有关详细信息，请参见 ItemPointerCompare。
 * ------------------------------------------------------------------------
 */
static int32 fc_tidcmp(const void *fc_a, const void *fc_b)
{
	ItemPointer fc_iptr1 = ((const ItemPointer) fc_a);
	ItemPointer fc_iptr2 = ((const ItemPointer) fc_b);

	return ItemPointerCompare(fc_iptr1, fc_iptr2);
}

/*-------------------------------------------------------------------------
 * sanity_check_tid_array()
 *
 * 对给定的 tid 数组执行有效性检查，并将 *ntids 设置为数组中项目的数量。
 * ------------------------------------------------------------------------
 */
static void fc_sanity_check_tid_array(ArrayType *fc_ta, int *fc_ntids)
{
	if (ARR_HASNULL(fc_ta) && array_contains_nulls(fc_ta))
		ereport(ERROR,
				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
				 errmsg("array must not contain nulls")));

	if (ARR_NDIM(fc_ta) > 1)
		ereport(ERROR,
				(errcode(ERRCODE_DATA_EXCEPTION),
				 errmsg("argument must be empty or one-dimensional array")));

	*fc_ntids = ArrayGetNItems(ARR_NDIM(fc_ta), ARR_DIMS(fc_ta));
}

/*-------------------------------------------------------------------------
 * find_tids_one_page()
 *
 * 找到与 tids[next_start_ptr] 在同一页面中的所有 tids，并更新 next_start_ptr 使其指向下一页面中的第一个 tid。
 *
 * 注意：输入的 tids[] 数组必须是已排序的。
 * ------------------------------------------------------------------------
 */
static BlockNumber fc_find_tids_one_page(ItemPointer fc_tids, int fc_ntids, OffsetNumber *fc_next_start_ptr)
{
	int			fc_i;
	BlockNumber fc_prev_blkno,
				fc_blkno;

	fc_prev_blkno = fc_blkno = InvalidBlockNumber;

	for (fc_i = *fc_next_start_ptr; fc_i < fc_ntids; fc_i++)
	{
		ItemPointerData fc_tid = fc_tids[fc_i];

		fc_blkno = ItemPointerGetBlockNumberNoCheck(&fc_tid);

		if (fc_i == *fc_next_start_ptr)
			fc_prev_blkno = fc_blkno;

		if (fc_prev_blkno != fc_blkno)
			break;
	}

	*fc_next_start_ptr = fc_i;
	return fc_prev_blkno;
}
